package net.qa.demo.rocketmq;

import java.util.List;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MessageProducer {
	private static final Logger log = LoggerFactory.getLogger(MessageProducer.class);
	
	static DefaultMQProducer producer;
	
	@BeforeClass
	public static void setUpBeforeClass() throws Exception {
		//构造消息生产者
        DefaultMQProducer defProducer = new DefaultMQProducer(MQConfig.GROUP);
        defProducer.setNamesrvAddr(MQConfig.NAME_SERVICE_ADDR);
        defProducer.setInstanceName(MQConfig.INSTANCE);
        defProducer.start();
        producer = defProducer;
	}

	@AfterClass
	public static void tearDownAfterClass() throws Exception {
	}

	@Before
	public void setUp() throws Exception {
	}

	@After
	public void tearDown() throws Exception {
	}

	@Test
	public void test() throws Exception {
		
		String msgContent = "Reliable synchronous transmission"; 
		for (int i=0; i<10; i++) {
	        Message msg = new Message(MQConfig.TOPIC, MQConfig.TAG,
	        		msgContent.getBytes(RemotingHelper.DEFAULT_CHARSET));
	        
	        //延时
	        msg.setDelayTimeLevel(3);
	        
	        //SendResult sendResult = producer.send(msg);
	        
	        SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
				
				@Override
				public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
					Integer index = (Integer) arg;
					System.out.println(index % mqs.size());
					return mqs.get(index % mqs.size());
				}
				
			}, i);
	        
	        log.info("{}\n{}", msgContent, sendResult);
	        
	        Thread.sleep(1000);
		}
		
	}

}
